package com.bw.gmall.realtime.app.dws;

import com.bw.gmall.realtime.utils.MyKafkaUtil;
import com.bw.gmall.realtime.utils.MysqlUtil;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwsTrafficShopDataWindow {
    public static void main(String[] args) {
        //todo 1.流环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //todo 2.设置并行度
        env.setParallelism(1);

        //表环境
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
        String topic ="dwd_trade_pay_detail_suc";
        String groupId ="DwsShopWindow";

        //创建表
        tenv.executeSql("CREATE TABLE Dws_shop_window (\n" +
                "  `user_id` BIGINT,\n" +
                "  `item_id` BIGINT,\n" +
                "  `behavior` STRING,\n" +
                "  `ts` TIMESTAMP(3) METADATA FROM 'timestamp'\n" +
                ") "+MyKafkaUtil.getKafkaDDL(topic,groupId));

    }
}
